package dht.paxos;
import dht.*;
import dht.message.*;
import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Paxos {
    
    private Lock lock = new ReentrantLock();
    
    private static final int SOCKET_TIMEOUT = 2000; // ms
    
    private final List<HostAddress> group;
    private final List<HostAddress> acceptors;
    private final List<OutstandingRequest> outstandingMessages;
    private Semaphore semaphore;
    
    private final Server node;
    private Star currentProposal;
    private PrepareRequestMessage lastSentProposal;
    
    private PrepareRequestMessage lastAcceptedProposal;
    
    private volatile boolean succeeded = false;
    
    private volatile State state;
    private void setState(State s) {
        this.state = s;
    }
    
    public Paxos (Server node) {
        try {
            this.group = Configuration.getAllHosts(node.getLocalAddress());
            System.out.println("Starting paxos with " + group.size() + " nodes");
        } catch (UnknownHostException ex) {
            throw new RuntimeException(ex);
        }
        
        this.node = node;
        this.semaphore = new Semaphore(0);
        this.acceptors = new CopyOnWriteArrayList<>();
        this.outstandingMessages = new CopyOnWriteArrayList<>();
        this.state = new AcceptorState();
    }
    
    public void messageReceived(PaxosMessage message) {
        OutstandingRequest outstanding = null;
        for (OutstandingRequest or : outstandingMessages) {
            if (or.isMatchingResponse(message)) {
                outstanding = or;
                break;
            }
        }
        
        if (outstanding != null) {
            System.out.println("removed it; " + message);
            outstandingMessages.remove(outstanding); 
        }
            
        state.messageReceived(message);
    }
    
    public void hostDied(HostAddress addr) {
        List<OutstandingRequest> requests = new ArrayList<OutstandingRequest>();
        for (OutstandingRequest request : outstandingMessages) {
            if (request.getAddress().equals(addr))
                requests.add(request);
        }
        
        outstandingMessages.removeAll(requests);
        group.remove(addr);
        acceptors.remove(addr);
        System.out.println("host " + addr.address() + " died.");
        if (acceptors.size() < 1) {
            System.out.println("no more acceptors ... releasing the blocking thread");
            setState(new AcceptorState());
            semaphore.release();
        }
    }
    
    private int getNumberOfHostsInGroup() {
        return group.size();
    }
    
    /**
     * used to add value into the dht atomically across all nodes
     * @param value
     * @return 
     */
    public boolean atomicBroadcast(Star value) throws InterruptedException {
        System.out.println("about to start new round of paxos ... ");
        succeeded = false;
        state = new ProposerState();
        state.start(value);
        semaphore.acquire();
        System.out.println("done with that round of paxos ... returning result " + succeeded);
        return succeeded;
    }
    
    private PrepareRequestMessage createNextProposal(HostAddress recipient) {
        lock.lock();
        try {
            lastSentProposal = new PrepareRequestMessage(getNextProposalNumber(), recipient);
            return lastSentProposal;
        } finally {
            lock.unlock();
        }
    }
    
    public boolean shouldResend(Message m) {
        // if the message is still an outstanding request, resend it
        for (OutstandingRequest request : outstandingMessages) 
            if (request.isSameRequest(m)) 
                return true;
        return false;
    }
    
    private int getNextProposalNumber() {
        lock.lock();
        try {
            if (lastSentProposal == null) {
                if (lastAcceptedProposal != null)
                    return lastAcceptedProposal.getProposalNumber() + 1;
                return getProposalNumberSeed();
            }
            else {
                if (lastAcceptedProposal != null)
                    return Math.max(lastAcceptedProposal.getProposalNumber(), lastSentProposal.getProposalNumber());
                return lastSentProposal.getProposalNumber() + 1;
            }
        } finally {
            lock.unlock();
        }
    }
    
    private int getProposalNumberSeed() {
        return 12345;
        // return ((ByteBuffer)ByteBuffer.wrap(addr.getAddress()).flip()).getInt();
    }
    
    private List<PrepareRequestMessage> createProposals(int proposalNum, List<HostAddress> hosts) {
        List<PrepareRequestMessage> proposals = new ArrayList<>();
        for (HostAddress host : hosts) 
            proposals.add(new PrepareRequestMessage(proposalNum, host));
        return proposals;
    }
    
    private List<AcceptRequestMessage> createAcceptRequests(int proposalNum, Star s, List<HostAddress> hosts) {
        List<AcceptRequestMessage> requests = new ArrayList<>();
        for (HostAddress host : hosts)
            requests.add(new AcceptRequestMessage(proposalNum, host, s));
        return requests;
    }
    
    private abstract class State {
        public void start(Star value) {
            throw new IllegalStateException();
        }
        
        public void messageReceived(PaxosMessage message) {
            throw new IllegalStateException();
        }
    }
    
    private final class AcceptorState extends State {
        @Override
        public void messageReceived(PaxosMessage message) {
            switch (message.getOpCode()) {
                case PrepareRequest :
                    prepareRequestReceived((PrepareRequestMessage)message);
                    break;
                case AcceptRequest :
                    acceptRequestReceived((AcceptRequestMessage)message);
                    break;
                default : break;
            }
        }
        
        private void prepareRequestReceived(PrepareRequestMessage message) {
            lock.lock();
            try {
                if(shouldAcceptProposal(message)) {
                    lastAcceptedProposal = message;
                    PrepareResponseMessage response = new PrepareResponseMessage(lastAcceptedProposal.getProposalNumber(), lastAcceptedProposal.getAddress());
                    node.sendMessage(response);
                } else {
                    // let the node know that they are sending prepare requests with lower numbers that what we've seen.
                    // send them the last accepted number + 1 since that is the least we will accept
                    PrepareResponseMessage response = new PrepareResponseMessage(lastAcceptedProposal.getProposalNumber() + 1, message.getAddress());
                    node.sendMessage(response);
                }
            } finally {
                lock.unlock();
            }
        }
        
        private boolean shouldAcceptProposal(PrepareRequestMessage proposal) {
            lock.lock();
            try {
                return (lastAcceptedProposal == null 
                        || proposal.getProposalNumber() > lastAcceptedProposal.getProposalNumber());
            } finally {
                lock.unlock();
            }
        }
        
        private void acceptRequestReceived(AcceptRequestMessage message) {
            if (shouldAcceptAcceptRequest(message)) {
                System.out.println("I agree on adding this value: " + message.getValue().toString());
                AcceptResponseMessage accept = new AcceptResponseMessage(message.getProposalNumber(), message.getAddress(), message.getValue());
                node.updateDatabase(message.getValue());
                node.sendMessage(accept);
            }
        }
        
        private boolean shouldAcceptAcceptRequest(AcceptRequestMessage message) {
            lock.lock();
            try {
                return lastAcceptedProposal != null
                    && lastAcceptedProposal.getAddress().equals(message.getAddress())
                    && message.getProposalNumber() >= message.getProposalNumber();

            } finally {
                lock.unlock();
            }
        }
    }
    
    private final class ProposerState extends State {
        @Override
        public void start(Star value) {
            lock.lock();
            try { currentProposal = value; } finally { lock.unlock(); }
            acceptors.clear();
            PrepareRequestMessage proposal = createNextProposal(null);
            List<PrepareRequestMessage> requests = createProposals(proposal.getProposalNumber(), group);
            System.out.println(node.getLocalAddress().hostName() + " is initiating paxos with proposal number " + proposal.getProposalNumber() + " -- and " + group.size() + " nodes ");
            for (PrepareRequestMessage request : requests) {
                outstandingMessages.add(new OutstandingPrepareRequest(request));
                node.sendReliably(request);
            }
        }
        
        @Override
        public void messageReceived(PaxosMessage message) {
            switch (message.getOpCode()) {
                case PrepareResponse :
                    prepareResponseReceived((PrepareResponseMessage)message);
                    break;
                case AcceptResponse :
                    acceptResponseReceived((AcceptResponseMessage)message);
                default : break;
            }
        }
        
        private void prepareResponseReceived(PrepareResponseMessage message) {
            lock.lock();
            try {
                System.out.println("Received " + message.getOpCode() + " from " + message.getAddress().hostName() + " with proposal number " + message.getProposalNumber());
                if (message.getProposalNumber() == lastSentProposal.getProposalNumber()) 
                    acceptors.add(message.getAddress());
                else if (message.getProposalNumber() > lastSentProposal.getProposalNumber()) {
                    // a proposal higher than ours has been made before this round ... restart the round.
                    System.out.println("\n\na higher proposal has already been made. restart with proposal: " + message.getProposalNumber() + "\n\n");
                    lastSentProposal = new PrepareRequestMessage(message.getProposalNumber(), message.getAddress());
                    outstandingMessages.clear();
                    acceptors.clear();
                    setState(new ProposerState());
                    semaphore.release();
                    return;
                }

                if (acceptors.size() > getNumberOfHostsInGroup()/2) {
                    System.out.println("Received a majority of responses");

                    AcceptRequestMessage ar = new AcceptRequestMessage(lastSentProposal.getProposalNumber(), null, currentProposal);
                    List<AcceptRequestMessage> requests = createAcceptRequests(ar.getProposalNumber(), ar.getValue(), acceptors);
                    System.out.println("sending " + ar.getOpCode() + " to all " + acceptors.size() + " acceptors with proposal number " + ar.getProposalNumber());
                    for (AcceptRequestMessage m : requests) {
                        outstandingMessages.add(new OutstandingAcceptRequest(m));
                        node.sendReliably(m);
                    }
                }
            } finally {
                lock.unlock();
            }
        }
        
        private void acceptResponseReceived(AcceptResponseMessage m) {
            System.out.println("received an " + m.getOpCode() + " from " + m.getAddress().hostName() + " with proposal number " + m.getProposalNumber());
            if (acceptors.contains(m.getAddress())) {
                acceptors.remove(m.getAddress());
            }
            System.out.println("there are  " + acceptors.size() + " acceptors that still need to respond");
            if (acceptors.size() < 1) {
                // we have gotten a response back from every acceptor saying that they added
                // the new value to their database
                setState(new AcceptorState());
                System.out.println("we are done!");
                System.out.println("only " + group.size() + " hosts are considered still alive");
                succeeded = true;
                semaphore.release();
            }
        }
    }
}
